package org.jie.guider.framework;

import java.util.concurrent.TimeUnit;
import org.jie.guider.framework.springboot.kafka.ChannelKafkaListenerEndpoint;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.config.KafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerEndpoint;
import org.springframework.kafka.config.KafkaListenerEndpointRegistry;
import org.springframework.kafka.listener.MessageListenerContainer;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

/**
 * @author LiJie
 */
@RestController
@RequestMapping("kafka")
public class KafkaController {
    private static final Logger LOG = LoggerFactory.getLogger(KafkaController.class);

    @Autowired
    private KafkaListenerEndpointRegistry kafkaRegistry;
    @Autowired
    private KafkaListenerContainerFactory<?> kafkaContainerFactory;

    @GetMapping("status")
    public String statusSwitch(@RequestParam String id,
                               @RequestParam String status) {

        MessageListenerContainer listener =
            kafkaRegistry.getListenerContainer(id);

        if (listener == null) {
            return "consumer not found.";
        }
        switch (status) {
            case "pause":
                listener.pause();
                break;
            case "stop":
                listener.stop();
                break;
            case "resume":
                listener.resume();
                break;
            case "start":
                listener.start();
                break;
            default:
                return "isPauseRequested: " + listener.isPauseRequested() +
                    "\nisContainerPaused: " + listener.isContainerPaused();
        }
        return "ok";
    }

    @GetMapping("create")
    public String registry(@RequestParam Long channel) {
        if (channel == null) {
            return "fail";
        }
        KafkaListenerEndpoint endpoint =
            new ChannelKafkaListenerEndpoint(channel, null);
        kafkaRegistry.registerListenerContainer(endpoint, kafkaContainerFactory);
        return "ok";
    }

    @GetMapping("exit")
    public void shutdown() {
        System.exit(0);
    }

    @KafkaListener(id = "guider-2",
                   groupId = "guider-group",
                   topics = "guider_test_topic")
    public void consumer2(String record) throws InterruptedException {
        LOG.info("[guider-2]消费数据: {}", record);
        TimeUnit.SECONDS.sleep(10);
    }
}
